/**
 * 2024/01/01 - 2024/12/31
 **/
package com.snail.starter.utils.queue.consumer;

import lombok.extern.slf4j.Slf4j;

import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.TimeUnit;
import java.util.function.Consumer;

/**
 * @author: (zg)
 * @Date: 2024/6/3 09:51
 * @Version: 1.0.0
 */
@Slf4j
public class QueueConsumer<T> extends Thread {

    private final ArrayBlockingQueue<T> blockingQueue;

    private final Consumer<T> consumer;

    private final long pollTimeout;

    private final TimeUnit pollTimeoutTimeUnit;

    private volatile boolean isRunning = true;

    public QueueConsumer(ArrayBlockingQueue<T> blockingQueue, Consumer<T> consumer) {
        this.blockingQueue = blockingQueue;
        this.consumer = consumer;
        this.pollTimeout = 1000L;
        this.pollTimeoutTimeUnit = TimeUnit.SECONDS;
    }

    public QueueConsumer(ArrayBlockingQueue<T> blockingQueue, Consumer<T> consumer, long pollTimeout, TimeUnit pollTimeoutTimeUnit) {
        this.blockingQueue = blockingQueue;
        this.consumer = consumer;
        this.pollTimeout = pollTimeout;
        this.pollTimeoutTimeUnit = pollTimeoutTimeUnit;
    }

    @Override
    public void run() {
        while (isRunning) {
            T t;
            try {
                t = blockingQueue.poll(pollTimeout, pollTimeoutTimeUnit);
                consumer.accept(t);
            } catch (InterruptedException e) {
                log.warn("================ InterruptedException poll time out !!! ================");
            } catch (Throwable throwable) {
                log.warn("", throwable);
                log.warn("================ Throwable error !!! ================");
            }
        }
    }

    public void stopConsumer() {
        this.isRunning = false;
    }

}